﻿#if NET451||NETSTANDARD2_0
using RabbitMQ.Client;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;

namespace RabbitMQExtension
{
    public partial class MQHelper : IDisposable
    {
        /// <summary> 保存MQ连接字典 </summary>
        private static ConcurrentDictionary<string, IModel> _modelDic = new ConcurrentDictionary<string, IModel>();
        /// <summary> MQ基础连接 </summary>
        private static IConnection _conn;
        /// <summary> 锁 </summary>
        private static readonly object _objLock = new object();
        /// <summary> 最后一次重新连接事件 </summary>
        private static DateTime _lastRecoveryTime = DateTime.Now;
        /// <summary> 重新连接的次数 </summary>
        private static int _retryCount = 0;
        /// <summary> MQ是否可用，true可用，false不可用 </summary>
        private static bool _available { get; set; } = false;

        /// <summary>
        /// 日记打印Log，可以外部传入接管默认Log
        /// </summary>
        public static ILog Log { get; set; } = LogManager.GetLogger(typeof(MQHelper));

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="config">配置</param>
        public MQHelper(MQConfig config)
        {
            if (string.IsNullOrEmpty(config.Host))
                throw new ArgumentNullException(nameof(config.Host));
            if (string.IsNullOrEmpty(config.UserName))
                throw new ArgumentNullException(nameof(config.UserName));
            if (string.IsNullOrEmpty(config.Password))
                throw new ArgumentNullException(nameof(config.Password));
            Open(config);
        }

        /// <summary>
        /// 创建链接
        /// </summary>
        /// <param name="config">配置</param>
        private static void Open(MQConfig config)
        {
            try
            {
                if (_conn != null) return;
                lock (_objLock)
                {
                    if (_conn != null) return;
                    var factory = new ConnectionFactory
                    {
                        UserName = config.UserName,
                        Password = config.Password,
                        VirtualHost = config.VirtualHost,
                        AutomaticRecoveryEnabled = true,
                        NetworkRecoveryInterval = TimeSpan.FromMilliseconds(10),
                    };
                    var amqps = new List<AmqpTcpEndpoint>();
                    var items = config.Host.Split(new char[] { ',', ';' }, StringSplitOptions.RemoveEmptyEntries);
                    foreach (var item in items)
                    {
                        var hosts = item.Split(':');
                        var port = 5672;
                        if (hosts.Length > 1)
                            int.TryParse(hosts[1], out port);
                        if (port < 1)
                            port = 5672;
                        amqps.Add(new AmqpTcpEndpoint
                        {
                            HostName = hosts[0],
                            Port = port,
                        });
                    }
                    _conn = _conn ?? factory.CreateConnection(amqps);
                    _conn.ConnectionShutdown += ConnectionShutdown;
                    _conn.ConnectionRecoveryError += ConnectionRecoveryError;
                    _conn.RecoverySucceeded += RecoverySucceeded;
                    _available = true;
                }
                Log.Info("MQ初始化成功");
            }
            catch (Exception e)
            {
                Log.Error("实例化MQ链接异常", e);
                _available = false;
            }
        }

        private static void RecoverySucceeded(object sender, EventArgs e)
        {
            _available = true;
            _retryCount = 0;
            Log.Info($"MQ 连接成功，断开时间:{_lastRecoveryTime},MQ 状态:{(_available ? "可用" : "不可用")}");
        }

        private static void ConnectionRecoveryError(object sender, RabbitMQ.Client.Events.ConnectionRecoveryErrorEventArgs e)
        {
            _retryCount++;
            Log.Info($"MQ 第 {_retryCount}次连接，断开时间:{_lastRecoveryTime},MQ 状态:{(_available ? "可用" : "不可用")}");
        }

        private static void ConnectionShutdown(object sender, ShutdownEventArgs e)
        {
            _lastRecoveryTime = DateTime.Now;
            _available = false;
            Log.Error($"MQ 断开连接，断开时间:{_lastRecoveryTime},MQ 状态:{(_available ? "可用" : "不可用")}");
        }

        /// <summary>
        /// 定义交换机
        /// </summary>
        /// <param name="exchange">交换机名称</param>
        /// <param name="type">交换机类型<see cref="ExchangeType"/></param>
        /// <param name="durable">是否持久化</param>
        /// <param name="autoDelete">是否自动删除</param>
        /// <param name="arguments">参数</param>
        private static void ExchangeDeclare(IModel model, string exchange, string type = ExchangeType.Direct, bool durable = true, bool autoDelete = false, IDictionary<string, object> arguments = null)
        {
            exchange = string.IsNullOrEmpty(exchange) ? "" : exchange.Trim();
            model.ExchangeDeclare(exchange, type, durable, autoDelete, arguments);
        }

        /// <summary>
        /// 定义队列
        /// </summary>
        /// <param name="queue">队列名称</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="exclusive">是否排外的，有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除；二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常</param>
        /// <param name="autoDelete">自动删除</param>
        /// <param name="arguments">参数</param>
        private static void QueueDeclare(IModel channel, string queue, bool durable = true, bool exclusive = false, bool autoDelete = false, IDictionary<string, object> arguments = null)
        {
            queue = string.IsNullOrEmpty(queue) ? "UndefinedQueue" : queue.Trim();
            channel.QueueDeclare(queue, durable, exclusive, autoDelete, arguments);
        }

        /// <summary>
        /// 获取Model
        /// </summary>
        /// <param name="exchange">交换机名称</param>
        /// <param name="queue">队列名称</param>
        /// <param name="routingKey">路由</param>
        /// <param name="durable">是否持久化</param>
        /// <returns></returns>
        private static IModel GetModel(string exchange, string queue, string routingKey, bool durable = true, string type = ExchangeType.Fanout)
        {
            return _modelDic.GetOrAdd(queue, key =>
            {
                var model = _conn.CreateModel();
                ExchangeDeclare(model, exchange, type, durable);
                QueueDeclare(model, queue, durable);
                model.QueueBind(queue, exchange, routingKey);
                _modelDic[queue] = model;
                return model;
            });
        }

        /// <summary>
        /// 发布消息
        /// </summary>
        /// <param name="exchange">交换机</param>
        /// <param name="queue">队列</param>
        /// <param name="routingKey">路由</param>
        /// <param name="body">消息内容</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="type">主题类型 <see cref="MQExchangeType"/>，默认为 <see cref="MQExchangeType.Fanout"/></param>
        public void Publish(string exchange, string queue, string routingKey, string body, bool durable = true, MQExchangeType type = MQExchangeType.Fanout)
        {
            var typeStr = GetExchangeTypeString(type);
            var channel = GetModel(exchange, queue, routingKey, durable, typeStr);
            lock (_objLock)
                channel.BasicPublish(exchange, routingKey, null, Encoding.UTF8.GetBytes(body));
        }

        /// <summary>
        /// 拉取消息
        /// </summary>
        /// <param name="exchange">交换机</param>
        /// <param name="queue">队列</param>
        /// <param name="routingKey">路由</param>
        /// <returns></returns>
        public string Pull(string exchange, string queue, string routingKey, bool durable = true, MQExchangeType type = MQExchangeType.Fanout)
        {
            var typeStr = GetExchangeTypeString(type);
            var channel = GetModel(exchange, queue, routingKey, durable, typeStr);

            try
            {
                lock (_objLock)
                {
                    // 从队列中获取一条消息
                    var result = channel.BasicGet(queue, false);

                    if (result == null)
                        return null;
                    string body = Encoding.UTF8.GetString(result.Body);
                    // 确认Id，MQ会将此条消息删除
                    channel.BasicAck(result.DeliveryTag, false);

                    return body;
                }
            }
            catch (Exception e)
            {
                Log.Error("MQ获取消息异常", e);
                if (e.Message.Contains("NOT_FOUND"))
                {
                    _modelDic.TryRemove(queue, out var val);
                    return Pull(exchange, queue, routingKey);
                }
                throw e;
            }
        }

        /// <summary>
        /// 获取队列中数据的数量
        /// </summary>
        /// <param name="exchange">交换机</param>
        /// <param name="queue">队列</param>
        /// <param name="routingKey">路由</param>
        /// <returns></returns>
        public uint GetMessageCount(string exchange, string queue, string routingKey, bool durable = true, MQExchangeType type = MQExchangeType.Fanout)
        {
            var typeStr = GetExchangeTypeString(type);
            var channel = GetModel(exchange, queue, routingKey, durable, typeStr);
            return channel.MessageCount(queue);
        }

        /// <summary> 
        /// MQ是否可用，true可用，false不可用
        /// </summary>
        public bool Available() => _available;

        /// <summary>
        /// 根据枚举获取主题类型
        /// </summary>
        /// <param name="type">交换机类型枚举</param>
        /// <returns></returns>
        private string GetExchangeTypeString(MQExchangeType type)
        {
            string typeStr;
            switch (type)
            {
                case MQExchangeType.Direct:
                    typeStr = ExchangeType.Direct;
                    break;
                case MQExchangeType.Topic:
                    typeStr = ExchangeType.Topic;
                    break;
                case MQExchangeType.Headers:
                    typeStr = ExchangeType.Headers;
                    break;
                default:
                    typeStr = ExchangeType.Fanout;
                    break;
            }
            return typeStr;
        }

        /// <summary>
        /// 释放链接
        /// </summary>
        public void Dispose()
        {
            foreach (var item in _modelDic)
            {
                item.Value.Dispose();
            }
            _conn.Dispose();
        }
    }
}

#endif